package org.example.domain.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.util.concurrent.AtomicDouble;
import jakarta.annotation.Resource;
import org.example.domain.service.OpenRankServiceDomain;
import org.example.domain.service.RepositoryServiceDomain;
import org.example.entity.openrank.DataNode;
import org.example.entity.openrank.MetaInfo;
import org.example.enums.FormulaCoefficientEnum;
import org.example.infra.service.IotDbMybatisService;
import org.example.vo.CommunityOpenRankVO;
import org.example.vo.ContributorVO;
import org.example.vo.KeyContributorsVO;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.time.Instant;
import java.time.ZoneOffset;
import java.time.ZonedDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * Description: RepositoryServiceDomain 服务实现类
 * Created by 陈宇 on 2024-12-27 14:37
 * Created with IntelliJ IDEA.
 * @author 陈宇
 */
@Service
public class RepositoryServiceDomainImpl implements RepositoryServiceDomain {

    @Resource(name = "mapCacheManager")
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Object>> cacheManager;
    @Resource
    private OpenRankServiceDomain openRankServiceDomain;
    @Resource
    private IotDbMybatisService iotDbMybatisService;
    @Resource
    private ExecutorService appThreadPool;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private RestTemplate httpRest;

    private List<String> months = new ArrayList<>(){{add("01");add("02");add("03");add("04");add("05");add("06");add("07");add("08");add("09");add("10");add("11");add("12");}};
    private Map<String, Integer> monthMap = new HashMap<>(){{put("01", 1);put("02",2);put("03",3);put("04",4);put("05",5);put("06",6);put("07",7);put("08",8);put("09",9);put("10",10);put("11",11);put("12",12);}};
    private Map<Integer, String> monthMap2Str = new HashMap<>(){{put(1, "01");put(2,"02");put(3,"03");put(4,"04");put(5,"05");put(6, "06");put(7,"07");put(8, "08");put(9,"09");put(10,"10");put(11,"11");put(12,"12");}};

    private Map<String, Map<String, String>> allCooperationMap = new HashMap<>();

    @Override
    public ContributorVO queryContributorByDate(String platform, String repoName, String userId, String startDate, String date) throws ExecutionException, InterruptedException {
        ContributorVO res = new ContributorVO();
        // 根据月份获取
        KeyContributorsVO keyContributorsVO = rangeQueryKeyContributors(repoName, platform, startDate, date);
        // 取出其中贡献者列表
        List<ContributorVO> contributorList = keyContributorsVO.getContributorList();
        // 遍历，找到目标用户
        for (ContributorVO contributorVO : contributorList) {
            if (contributorVO.getUserId().equals(userId)) {
                res = contributorVO;
                break;
            }
        }
        return res;
    }

    @Override
    public KeyContributorsVO rangeQueryKeyContributors(String name, String platform, String startDate, String date) throws ExecutionException, InterruptedException {

        StringBuilder sb = new StringBuilder();
        sb.append(platform).append("/").append(name);
        ArrayList<ContributorVO> contributorVOS = new ArrayList<>();
        KeyContributorsVO keyContributorsVO = new KeyContributorsVO();

        String[] startYearAndMonth = startDate.split("-");
        String[] endYearAndMonth = date.split("-");
        // 起始年份
        String startYear = startYearAndMonth[0];
        // 起始月份
        String startMonth = startYearAndMonth[1];
        // 结束年份
        String endYear = endYearAndMonth[0];
        // 结束月份
        String endMonth = endYearAndMonth[1];
        // 关键贡献者人数
        int num = 0;

        HashMap<String, String> starMap = new HashMap<>();
        HashMap<String, String> forkMap = new HashMap<>();
        HashMap<String, String> prMap = new HashMap<>();
        HashMap<String, String> issueMap = new HashMap<>();

        CompletableFuture<Void> completableFuture1 = null;
        CompletableFuture<Void> completableFuture2 = null;

        while (!startYear.equals(String.valueOf(Integer.valueOf(endYear)+1)) || !startMonth.equals(endMonth)) {
            if (startYear.equals(endYear)) {
                // 年份相同，月份不同
                Integer startMonthInt = monthMap.get(startMonth);
                Integer endMonthInt = monthMap.get(endMonth);
                while (startMonthInt <= endMonthInt) {
                    Integer finalStartMonthInt = startMonthInt;
                    String finalStartYear = startYear;
                    completableFuture1 = CompletableFuture.runAsync(() -> {
                        // 获取时间
                        ZonedDateTime currentTime = ZonedDateTime.of(Integer.valueOf(finalStartYear), finalStartMonthInt, 1, 0, 0, 0, 0, ZoneOffset.UTC);
                        // 获取 1970-01-01T00:00:00Z 对应的 Instant
                        Instant epoch = Instant.EPOCH;
                        // 下一个月的时间
                        ZonedDateTime nextTime = currentTime.plusMonths(1);
                        // 时间截止
                        long nextMillis = nextTime.toInstant().toEpochMilli() - epoch.toEpochMilli();
                        String repoId = openRankServiceDomain.getRepoId(platform, name);
                        if (repoId != null) {
                            // 获取一个月的 star 数、fork 数、issue 数、pr 数
                            Integer starCount = iotDbMybatisService.getStarCount(repoId, nextMillis);
                            Integer forkCount = iotDbMybatisService.getForkCount(repoId, nextMillis);
                            Integer issueCount = iotDbMybatisService.getIssueCount(repoId, nextMillis);
                            Integer prCount = iotDbMybatisService.getPrCount(repoId, nextMillis);
                            starMap.put(finalStartYear + "-" + monthMap2Str.get(finalStartMonthInt), starCount.toString());
                            forkMap.put(finalStartYear + "-" + monthMap2Str.get(finalStartMonthInt), forkCount.toString());
                            issueMap.put(finalStartYear + "-" + monthMap2Str.get(finalStartMonthInt), issueCount.toString());
                            prMap.put(finalStartYear + "-" + monthMap2Str.get(finalStartMonthInt), prCount.toString());
                        }
                    }, appThreadPool);
                    // 获取一个月的贡献者数据
                    KeyContributorsVO vo = queryKeyContributorList(name, platform, startYear + "-" + monthMap2Str.get(startMonthInt), 0);

                    if (startMonthInt.equals(endMonthInt)) {
                        // 去最后一个月的规模
                        num = vo.getNum();
                    }
                    // 取出其中的关键贡献者列表
                    List<ContributorVO> contributorList = vo.getContributorList();
                    // 遍历总表
                    List<ContributorVO> tmpList = new ArrayList<>(contributorList);
                    for (ContributorVO cVO : contributorVOS) {
                        for (ContributorVO cLVO : contributorList) {
                            if (cVO.getUserId().equals(cLVO.getUserId())) {
                                // 更新 openRank
                                double openRank = Double.valueOf(cVO.getOpenRank()) + Double.valueOf(cLVO.getOpenRank());
                                cVO.setOpenRank(String.valueOf(openRank));
                                // 更新 协作度
                                double cooperation = Double.valueOf(cVO.getCooperation()) + Double.valueOf(cLVO.getCooperation());
                                cVO.setCooperation(String.valueOf(cooperation));
                                // 更新 影响力
                                double effective = cVO.getEffective() + cLVO.getEffective();
                                cVO.setEffective(effective);
                                // 更新连接边
                                Map<String, String> cVOLink = cVO.getLink();
                                Map<String, String> cLVOLink = cLVO.getLink();
                                Set<String> key = cLVOLink.keySet();
                                for (String k : key) {
                                    String number = cVOLink.getOrDefault(k, "0");
                                    long value = Long.valueOf(number) + Long.valueOf(cLVOLink.get(k));
                                    cVOLink.put(k, String.valueOf(value));
                                }
                                cVO.setLink(cVOLink);
                                // 从 tmpList 中移除
                                tmpList.remove(cLVO);
                            }
                        }
                    }

                    contributorVOS.addAll(tmpList);
                    // 去往下一个月
                    startMonthInt++;
                }
                startMonth = endMonth;
                startYear = String.valueOf(Integer.valueOf(startYear)+1);
            } else {
                // 年份不同的时候，先把月份加到 12 月，然后再加年
                Integer startYearInt = Integer.valueOf(startYear);
                Integer startMonthInt = monthMap.get(startMonth);
                while (startMonthInt <= 12) {
                    String finalStartYear1 = startYear;
                    Integer finalStartMonthInt = startMonthInt;
                    completableFuture2 = CompletableFuture.runAsync(() -> {
                        // 获取时间
                        ZonedDateTime currentTime = ZonedDateTime.of(Integer.valueOf(finalStartYear1), finalStartMonthInt, 1, 0, 0, 0, 0, ZoneOffset.UTC);
                        // 获取 1970-01-01T00:00:00Z 对应的 Instant
                        Instant epoch = Instant.EPOCH;
                        // 下一个月的时间
                        ZonedDateTime nextTime = currentTime.plusMonths(1);
                        // 时间截止
                        long nextMillis = nextTime.toInstant().toEpochMilli() - epoch.toEpochMilli();
                        String repoId = openRankServiceDomain.getRepoId(platform, name);
                        if (repoId != null) {
                            // 获取一个月的 star 数、fork 数、issue 数、pr 数
                            Integer starCount = iotDbMybatisService.getStarCount(repoId, nextMillis);
                            Integer forkCount = iotDbMybatisService.getForkCount(repoId, nextMillis);
                            Integer issueCount = iotDbMybatisService.getIssueCount(repoId, nextMillis);
                            Integer prCount = iotDbMybatisService.getPrCount(repoId, nextMillis);
                            starMap.put(finalStartYear1 + "-" + monthMap2Str.get(finalStartMonthInt), starCount.toString());
                            forkMap.put(finalStartYear1 + "-" + monthMap2Str.get(finalStartMonthInt), forkCount.toString());
                            issueMap.put(finalStartYear1 + "-" + monthMap2Str.get(finalStartMonthInt), issueCount.toString());
                            prMap.put(finalStartYear1 + "-" + monthMap2Str.get(finalStartMonthInt), prCount.toString());
                        }
                    }, appThreadPool);
                    // 获取一个月的贡献者数据
                    KeyContributorsVO vo = queryKeyContributorList(name, platform, startYear + "-" + monthMap2Str.get(startMonthInt), 0);
                    // 取出其中的关键贡献者列表
                    List<ContributorVO> contributorList = vo.getContributorList();
                    // 遍历总表
                    List<ContributorVO> tmpList = new ArrayList<>(contributorList);
                    for (ContributorVO cVO : contributorVOS) {
                        for (ContributorVO cLVO : contributorList) {
                            if (cVO.getUserId().equals(cLVO.getUserId())) {
                                // 更新 openRank
                                double openRank = Double.valueOf(cVO.getOpenRank()) + Double.valueOf(cLVO.getOpenRank());
                                cVO.setOpenRank(String.valueOf(openRank));
                                // 更新 协作度
                                double cooperation = Double.valueOf(cVO.getCooperation()) + Double.valueOf(cLVO.getCooperation());
                                cVO.setCooperation(String.valueOf(cooperation));
                                // 更新 影响力
                                double effective = cVO.getEffective() + cLVO.getEffective();
                                cVO.setEffective(effective);
                                // 更新连接边
                                Map<String, String> cVOLink = cVO.getLink();
                                Map<String, String> cLVOLink = cLVO.getLink();
                                Set<String> key = cLVOLink.keySet();
                                for (String k : key) {
                                    String number = cVOLink.getOrDefault(k, "0");
                                    long value = Long.valueOf(number) + Long.valueOf(cLVOLink.get(k));
                                    cVOLink.put(k, String.valueOf(value));
                                }
                                cVO.setLink(cVOLink);
                                // 从 tmpList 中移除
                                tmpList.remove(cLVO);
                            }
                        }
                    }

                    contributorVOS.addAll(tmpList);
                    // 去往下一个月
                    startMonthInt++;
                }
                startMonth = "01";
                startYear = String.valueOf(startYearInt + 1);
            }
        }

        // 优先级队列 按照影响力 由大到小排序
        Queue<ContributorVO> contributorVOSQueue = new PriorityQueue<>(new Comparator<ContributorVO>() {
            @Override
            public int compare(ContributorVO o1, ContributorVO o2) {
                return Double.compare(o2.getEffective(), o1.getEffective());
            }
        });
        contributorVOSQueue.addAll(contributorVOS);
        ArrayList<ContributorVO> resContributorVO = new ArrayList<>();
        while (contributorVOSQueue.size() > 0) {
            resContributorVO.add(contributorVOSQueue.poll());
        }

        keyContributorsVO.setContributorList(resContributorVO);
        keyContributorsVO.setNum(num);

        // 分析地理数据
        // 分析用户地理位置信息
        // 遍历用户信息
        HashMap<String, Integer> map = new HashMap<>();
        for (ContributorVO contributorVO : resContributorVO) {
            if (num == 0) {
                break;
            }
            num--;
            String location = contributorVO.getLocation();
            if (location != null) {
                String[] split = location.split(",");
                String country = split[split.length - 1];
                country = country.trim();
                map.put(country, map.getOrDefault(country, 0)+1);
            } else {
                map.put("UnKnow", map.getOrDefault("UnKnow", 0)+1);
            }

        }
        HashMap<String, Double> res = new HashMap<>();
        int userCount = keyContributorsVO.getNum();
        map.forEach((k, v) -> {
            res.put(k, (double)v/userCount);
        });
        keyContributorsVO.setContributorLocation(res);

        // 往 allCooperationMap 中填充
        allCooperationMap.clear();
        List<ContributorVO> contributorList = keyContributorsVO.getContributorList();
        for (ContributorVO contributor : contributorList) {
            allCooperationMap.put(contributor.getUserId(), contributor.getLink());
        }
        // 对数据做清洗
        dataClarity(keyContributorsVO);

        if (completableFuture1 != null) {
            completableFuture1.get();
        }
        if (completableFuture2 != null) {
            completableFuture2.get();
        }
        keyContributorsVO.setStarMap(starMap);
        keyContributorsVO.setIssueMap(issueMap);
        keyContributorsVO.setPrMap(prMap);
        keyContributorsVO.setForkMap(forkMap);
        System.out.println("结果已返回");
        // 对数据做归一化？
        return keyContributorsVO;
    }

    @Override
    public KeyContributorsVO queryKeyContributorsByYear(String name, String platform, String date) throws ExecutionException, InterruptedException {
        KeyContributorsVO keyContributorsVO = new KeyContributorsVO();
        ArrayList<ContributorVO> contributorVOS = new ArrayList<>();


        StringBuilder sb = new StringBuilder();
        sb.append(platform).append("/").append(name);
        // 从 redis 中获取
        StringBuilder sbRedis = new StringBuilder();
        sbRedis.append("openSODA").append(":").append("contributor").append(":").append(sb.toString()).append(":").append(date);
        String resKeyContributorsVO = stringRedisTemplate.opsForValue().get(sbRedis.toString());
        if (resKeyContributorsVO != null) {
            // 有就从缓存中获取
            keyContributorsVO = JSONObject.parseObject(resKeyContributorsVO, KeyContributorsVO.class);
        } else {
            // 没有就计算
            // 获取年份
            String year = date;

            for (int i = 1; i <= 12; i++) {
                if (year.equals("2023")) {
                    if (i == 4) {
                        break;
                    }
                }
                System.out.println("正在统计【"+year+" 年 "+i + "月 】的数据");
                // 月份
                String month = monthMap2Str.get(i);
                KeyContributorsVO vo = queryKeyContributorList(name, platform, year + "-" + month, 0);
                // 取出其中的关键贡献者列表
                List<ContributorVO> contributorList = vo.getContributorList();
                // 遍历总表
                List<ContributorVO> tmpList = new ArrayList<>(contributorList);
                for (ContributorVO cVO : contributorVOS) {
                    for (ContributorVO cLVO : contributorList) {
                        if (cVO.getUserId().equals(cLVO.getUserId())) {
                            // 更新 openRank
                            double openRank = Double.valueOf(cVO.getOpenRank()) + Double.valueOf(cLVO.getOpenRank());
                            cVO.setOpenRank(String.valueOf(openRank));
                            // 更新 协作度
                            double cooperation = Double.valueOf(cVO.getCooperation()) + Double.valueOf(cLVO.getCooperation());
                            cVO.setCooperation(String.valueOf(cooperation));
                            // 更新 影响力
                            double effective = cVO.getEffective() + cLVO.getEffective();
                            cVO.setEffective(effective);
                            // 更新连接边
                            Map<String, String> cVOLink = cVO.getLink();
                            Map<String, String> cLVOLink = cLVO.getLink();
                            Set<String> key = cLVOLink.keySet();
                            for (String k : key) {
                                String number = cVOLink.getOrDefault(k, "0");
                                long value = Long.valueOf(number) + Long.valueOf(cLVOLink.get(k));
                                cVOLink.put(k, String.valueOf(value));
                            }
                            cVO.setLink(cVOLink);
                            // 从 tmpList 中移除
                            tmpList.remove(cLVO);
                        }
                    }
                }

                contributorVOS.addAll(tmpList);
            }

            // 优先级队列 按照影响力 由大到小排序
            Queue<ContributorVO> contributorVOSQueue = new PriorityQueue<>(new Comparator<ContributorVO>() {
                @Override
                public int compare(ContributorVO o1, ContributorVO o2) {
                    return Double.compare(o2.getEffective(), o1.getEffective());
                }
            });
            contributorVOSQueue.addAll(contributorVOS);
            ArrayList<ContributorVO> resContributorVO = new ArrayList<>();
            while (contributorVOSQueue.size() > 0) {
                resContributorVO.add(contributorVOSQueue.poll());
            }

            keyContributorsVO.setContributorList(resContributorVO);

            // 分析地理数据
            // 分析用户地理位置信息
            // 遍历用户信息
            HashMap<String, Integer> map = new HashMap<>();
            for (ContributorVO contributorVO : contributorVOS) {
                String location = contributorVO.getLocation();
                if (location != null) {
                    String[] split = location.split(",");
                    String country = split[split.length - 1];
                    country = country.trim();
                    map.put(country, map.getOrDefault(country, 0)+1);
                } else {
                    map.put("UnKnow", map.getOrDefault("UnKnow", 0)+1);
                }

            }
            HashMap<String, Double> res = new HashMap<>();
            int userCount = contributorVOS.size();
            map.forEach((k, v) -> {
                res.put(k, (double)v/userCount);
            });
            keyContributorsVO.setContributorLocation(res);
            try {
                // 存入缓存中
                stringRedisTemplate.opsForValue().set(sbRedis.toString(), JSON.toJSONString(keyContributorsVO));
            } catch (Exception e) {
                System.out.println("redis 存入【 " + date + " 】 结果失败");
                e.printStackTrace();
            }
        }
        // 往 allCooperationMap 中填充
        allCooperationMap.clear();
        List<ContributorVO> contributorList = keyContributorsVO.getContributorList();
        for (ContributorVO contributor : contributorList) {
            allCooperationMap.put(contributor.getUserId(), contributor.getLink());
        }
        // 数据清洗
        dataClarity(keyContributorsVO);
        System.out.println("成功获取 【 " + date + "年 】的关键贡献者");
        return keyContributorsVO;
    }

    @Override
    public KeyContributorsVO queryKeyContributorsByMonth(String name, String platform, String date, Integer type) throws ExecutionException, InterruptedException {
        System.out.println("开始获取【 " + date + " 】的数据");
        KeyContributorsVO keyContributorsVO = queryKeyContributorList(name, platform, date, type);
        // 往 allCooperationMap 中填充
        allCooperationMap.clear();
        List<ContributorVO> contributorList = keyContributorsVO.getContributorList();
        for (ContributorVO contributor : contributorList) {
            allCooperationMap.put(contributor.getUserId(), contributor.getLink());
        }
        // 对数据做清洗
        dataClarity(keyContributorsVO);
        System.out.println("【 " + date + " 】数据获取完成");

        return keyContributorsVO;
    }

    public KeyContributorsVO queryKeyContributorList(String name, String platform, String date, Integer type) throws ExecutionException, InterruptedException {

        KeyContributorsVO keyContributorsVO = new KeyContributorsVO();
        StringBuilder sb = new StringBuilder();
        sb.append(platform).append("/").append(name);
        // 先从缓存中查，没有就进入下面的逻辑
        StringBuilder sbRedis = new StringBuilder();
        sbRedis.append("openSODA").append(":").append("contributor").append(":").append(sb.toString()).append(":").append(date);
        String keyContributorsVOStr = stringRedisTemplate.opsForValue().get(sbRedis.toString());
        if (keyContributorsVOStr != null) {
            // 缓存中有就从缓存中获取
            keyContributorsVO = JSONObject.parseObject(keyContributorsVOStr, KeyContributorsVO.class);
        } else {
            // 缓存中没有就自己计算
            CommunityOpenRankVO communityOpenRankVO = openRankServiceDomain.queryCommunityOpenRank(platform, name);
            // 按月来查
            keyContributorsVO = queryKeyContributorListByMonth(sb.toString(), date, communityOpenRankVO.getMeta());
        }
        // 分析用户地理位置信息
        List<ContributorVO> contributorList = keyContributorsVO.getContributorList();
        // 获取关键贡献者数量
        Integer num = keyContributorsVO.getNum();
        List<ContributorVO> finalContributorList = new ArrayList<>();
        while (num > 0) {
            finalContributorList.add(contributorList.get(num-1));
            num--;
        }
        // 遍历用户信息
        HashMap<String, Integer> map = new HashMap<>();
        for (ContributorVO contributorVO : finalContributorList) {
            String location = contributorVO.getLocation();
            if (location != null) {
                String[] split = location.split(",");
                String country = split[split.length - 1];
                country = country.trim();
                map.put(country, map.getOrDefault(country, 0)+1);
            } else {
                map.put("UnKnow", map.getOrDefault("UnKnow", 0)+1);
            }

        }
        HashMap<String, Double> res = new HashMap<>();
        int userCount = finalContributorList.size();
        map.forEach((k, v) -> {
            res.put(k, (double)v/userCount);
        });
        keyContributorsVO.setContributorLocation(res);
        System.out.println("成功获取 【 " + date.split("-")[0] + " 年 " + date.split("-")[1] + " 月 】的关键贡献者");
        return keyContributorsVO;
    }

    /**
     * 对双向边连接数据做清洗
     */
    private void dataClarity (KeyContributorsVO keyContributorsVO) {
        // 获取未清洗数据的贡献者列表
        List<ContributorVO> contributorList = keyContributorsVO.getContributorList();
        // 协作是双向的关系，你评论了别人，别人不一定评论你
        // 获取一个用户的 link，从其中取出 key（评论他人的用户 id），再用这个 id 去判断 这个用户是是否和本用户有协作，取两者的最小 value
        for (ContributorVO contributor : contributorList) {
            // 评论他人的用户 id
            Map<String, String> link = contributor.getLink();
            Set<String> keySet = link.keySet();
            for (String key : keySet) {
                // 和该用户的协作次数
                Integer nums = Integer.valueOf(link.get(key));
                // 寻找该用户的 link
                Map<String, String> othersLink = allCooperationMap.get(key);
                Integer othersNum = Integer.valueOf(othersLink.getOrDefault(contributor.getUserId(), "0"));
                if (nums > othersNum) {
                    link.put(key, othersNum.toString());
                }
            }
            AtomicInteger num = new AtomicInteger(0);
            link.values().forEach( v -> {
                num.addAndGet(Integer.valueOf(v));
            });
            if (num.get() == 0) {
                contributor.setCooperation("0");
                contributor.setEffective(Double.valueOf(contributor.getOpenRank()));
            }
        }
    }

    /**
     * 按月来查
     * @param name
     * @param date
     * @return
     */
    private KeyContributorsVO queryKeyContributorListByMonth(String name, String date, MetaInfo metaInfo) throws ExecutionException, InterruptedException {
        allCooperationMap.clear();
        /**
         * 记录各个用户彼此协作的次数 hash 表
         */
        Map<String, Map<String, Integer>> cooperationPrNumMap = new HashMap<>();
        Map<String, Map<String, Integer>> cooperationIssueNumMap = new HashMap<>();

        KeyContributorsVO keyContributorsVO = new KeyContributorsVO();
        // 优先级队列 按照影响力 由大到小排序
        Queue<ContributorVO> contributorVOS = new PriorityQueue<>(new Comparator<ContributorVO>() {
            @Override
            public int compare(ContributorVO o1, ContributorVO o2) {
                return Double.compare(o2.getEffective(), o1.getEffective());
            }
        });
        String[] split = date.split("-");
        // 获取其中的年份
        String year = split[0];
        // 获取月份
        String month = split[1];

        // 上个月的年份
        String lastMonthYear = month.equals("01") ? String.valueOf((Integer.valueOf(year)-1)) : year;
        // 上一个月
        String lastMonth = monthMap2Str.getOrDefault(monthMap.get(month)-1, "12");

        // 获取仓库内这个月度活跃用户，每个用户都有其 openrank 值，即 贡献度
        ConcurrentHashMap<String, String> userOpenRankMap = new ConcurrentHashMap<>();
        // 上一个月度活跃用户的 openRank
        ConcurrentHashMap<String, String> lastMonthUserOpenRankMap = new ConcurrentHashMap<>();

        // 注入本月的活跃用户及其 openRank
        DataNode dataNode = (DataNode) ((Map<String, Object>)cacheManager.get("communityOpenRankCache")
                .get(name))
                .get(year + month);
        if (dataNode == null) {
            return keyContributorsVO;
        }
        List<List<Double>> nodes = dataNode.getNodes();
        nodes.stream().forEach(n -> {
            // 遍历所有节点，通过节点 id，在缓存中找到其 平台 id
            String platformId = (String) cacheManager.get("nodeId").get(n.get(0).toString());
            // 判断该 平台 id 是否是用户
            if (platformId != null) {
                if (platformId.charAt(0) == 'u') {
                    // 如果是用户则存入 userOpenRankMap
                    userOpenRankMap.put(platformId, n.get(2).toString());
                }
            }
            // 如果不是则跳过
        });
        // 注入上个月的活跃用户及其 openRank
        DataNode lastMonthDataNode = (DataNode) ((Map<String, Object>)cacheManager.get("communityOpenRankCache")
                .get(name))
                .get(lastMonthYear + lastMonth);
        List<List<Double>> lastMonthNodes = lastMonthDataNode.getNodes();
        lastMonthNodes.stream().forEach(n -> {
            // 遍历所有节点，通过节点 id，在缓存中找到其 平台 id
            String platformId = (String) cacheManager.get("nodeId").get(n.get(0).toString());
            // 判断该 平台 id 是否是用户
            if (platformId != null) {
                if (platformId.charAt(0) == 'u') {
                    // 如果是用户则存入 userOpenRankMap
                    lastMonthUserOpenRankMap.put(platformId, n.get(2).toString());
                }
            }
            // 如果不是则跳过
        });

        // 获取活跃用户在平台上的 id
        List<String> userList = userOpenRankMap.keySet().stream().map(key -> {
            String str = (String) key;
            str = str.substring(1, str.length());
            return str;
        }).collect(Collectors.toList());

        // 获取每个人的 pr 中的协作次数
        ZonedDateTime currentTime = ZonedDateTime.of(Integer.valueOf(split[0]), monthMap.get(split[1]), 1, 0, 0, 0, 0, ZoneOffset.UTC);
        // 获取 1970-01-01T00:00:00Z 对应的 Instant
        Instant epoch = Instant.EPOCH;
        // 时间差
        long currentMillis = currentTime.toInstant().toEpochMilli() - epoch.toEpochMilli();
        // 下一个月的时间
        ZonedDateTime nextTime = currentTime.plusMonths(1);
        // 时间截止
        long nextMillis = nextTime.toInstant().toEpochMilli() - epoch.toEpochMilli();
        // 计算活跃用户的 协作度
        // 测试处理到第几个用户
        AtomicInteger testNum = new AtomicInteger(-1);
        for (String userId : userList) {
            System.out.println("正在处理用户：【"+userId+"】的评论数据，已处理用户数：【"+testNum.incrementAndGet()+"】");
            // 获取所有 pr 评论
            Set<Long> prCommentIds = iotDbMybatisService.queryPRs(metaInfo.getRepoId(), userId, currentMillis, nextMillis).stream().map(data -> {
                return data.getPullReviewCommentId();
            }).collect(Collectors.toSet());
            Map<String, Map<String, Integer>> finalCooperationPrNumMap = cooperationPrNumMap;
            CompletableFuture<Void> prCommentFuture = CompletableFuture.runAsync(() -> {
                for (Long prCommentId : prCommentIds) {
                    // 查看每条评论内容
                    String body = iotDbMybatisService.queryPrCommentBody(prCommentId);
                    // 将字符串以空格分开，这样第一个就是 @
                    String[] content = body.split(" ");
                    if (content[0].contains("@")) {
                        // 说明这是一条回复评论, 截取被@的用户id
                        String cooperationUser = content[0].substring(1, content[0].length());
                        String cooperationUserId = "";
                        for (List<String> node : metaInfo.getNodes()) {
                            // 此处取得的是整个仓库所有用户
                            if (node.get(1).equals(cooperationUser)) {
                                // 不带 u 的
                                cooperationUserId = node.get(0).substring(1, node.get(0).length());
                                Map<String, Integer> numMap = finalCooperationPrNumMap.getOrDefault(userId, new HashMap<>());
                                finalCooperationPrNumMap.put(userId, numMap);
                                numMap.put(cooperationUserId, numMap.getOrDefault(cooperationUserId, 0) + 1);
                            }
                        }
                    }
                }

                // 做数据的处理，可能你评论了别人，但是别人没评论你
//                Set<String> keySet = finalCooperationPrNumMap.keySet();
//                for (String key : keySet) {
//                    Map<String, Integer> cooperationMap = finalCooperationPrNumMap.get(key);
//                    Set<String> cooperationIds = cooperationMap.keySet();
//                    for (String cooperationId : cooperationIds) {
//                        Map<String, Integer> map = finalCooperationPrNumMap.get(cooperationId);
//                        if (map != null) {
//                            // 取出发给这个用户的消息次数
//                            Integer action = cooperationMap.get(cooperationId);
//                            // 取出被发送消息的用户回复的次数
//                            Integer beAction = map.getOrDefault(key, 0);
//                            if (action > beAction) {
//                                cooperationMap.put(cooperationId, beAction);
//                            }
//                        } else {
//                            cooperationMap.put(cooperationId, 0);
//                        }
//                    }
//                }

            }, appThreadPool);
            // 获取所有 issue 评论
            Set<Long> issueCommentIds = iotDbMybatisService.queryIssues(metaInfo.getRepoId(), userId, currentMillis, nextMillis).stream().map(data -> {
                return data.getIssueCommentId();
            }).collect(Collectors.toSet());
            Map<String, Map<String, Integer>> finalCooperationIssueNumMap = cooperationIssueNumMap;
            CompletableFuture<Void> issueCommentFuture = CompletableFuture.runAsync(() -> {
                for (Long issueCommentId : issueCommentIds) {
                    String body = iotDbMybatisService.queryIssueCommentBody(issueCommentId);
                    // 将字符串以空格分开，这样第一个就是 @
                    String[] content = body.split(" ");
                    if (content[0].contains("@")) {
                        // 说明这是一条回复评论, 截取被@的用户id
                        String cooperationUser = content[0].substring(1, content[0].length());
                        String cooperationUserId = "";
                        for (List<String> node : metaInfo.getNodes()) {
                            if (node.get(1).equals(cooperationUser)) {
                                // 不带 u 的
                                cooperationUserId = node.get(0).substring(1, node.get(0).length());
                                Map<String, Integer> numMap = finalCooperationIssueNumMap.getOrDefault(userId, new HashMap<>());
                                finalCooperationIssueNumMap.put(userId, numMap);
                                numMap.put(cooperationUserId, numMap.getOrDefault(cooperationUserId, 0) + 1);
                            }
                        }
                    }
                }

                // 做数据的处理，可能你评论了别人，但是别人没评论你
//                Set<String> keySet = finalCooperationIssueNumMap.keySet();
//                for (String key : keySet) {
//                    Map<String, Integer> cooperationMap = finalCooperationIssueNumMap.get(key);
//                    Set<String> cooperationIds = cooperationMap.keySet();
//                    for (String cooperationId : cooperationIds) {
//                        Map<String, Integer> map = finalCooperationIssueNumMap.get(cooperationId);
//                        if (map != null) {
//                            // 取出发给这个用户的消息次数
//                            Integer action = cooperationMap.get(cooperationId);
//                            // 取出被发送消息的用户回复的次数
//                            Integer beAction = map.getOrDefault(key, 0);
//                            if (action > beAction) {
//                                cooperationMap.put(cooperationId, beAction);
//                            }
//                        } else {
//                            cooperationMap.put(cooperationId, 0);
//                        }
//                    }
//                }
            }, appThreadPool);
            prCommentFuture.get();
            issueCommentFuture.get();
        }

        List<String> finalUserList = userList;
        CompletableFuture<Void> cooperationFuture = CompletableFuture.runAsync(() -> {
            for (String userId : finalUserList) {
                // 用户 userId 的协作度
                // 协作用户 id，及次数
                HashMap<String, String> cooperationUserNum = new HashMap<>();
                Double cooperation = cooperationDegreeCalculation(userId, finalUserList, userOpenRankMap, lastMonthUserOpenRankMap, cooperationPrNumMap, cooperationIssueNumMap, cooperationUserNum);
                // 通过 贡献度 以及 协作度，计算这个月活跃用户的 影响力，并按照影响力从低到高排序
                String openRank = userOpenRankMap.get("u" + userId);
                ContributorVO contributorVO = new ContributorVO();
                contributorVO.setUserId(userId);
                // 根据 id 找到其 序号
                String oId = (String) cacheManager.get("nodeIdR").get("u" + userId);
                Integer oIdInt = Integer.valueOf(oId.substring(0, oId.length() - 2));
                // 去元数据中取 用户昵称
                List<List<String>> metaInfoNodes = metaInfo.getNodes();
                String location = null;
                if (oIdInt < metaInfoNodes.size()) {
                    String userName = metaInfoNodes.get(oIdInt).get(1);
                    contributorVO.setUserName(userName);
                    String url = "https://api.github.com/users/{1}";
                    HttpHeaders httpHeaders = new HttpHeaders();
                    // 获取 社区 OpenRank
                    try {
                        ResponseEntity<Map> exchange = httpRest.exchange(url, HttpMethod.GET, new HttpEntity<>(httpHeaders), Map.class, userName);
                        Map body = exchange.getBody();
                        location = (String) body.get("location");
                    } catch (Exception e) {
                        System.out.println("用户【"+userId+"】地理数 github api 无法解析");
                    }
                }
                contributorVO.setLocation(location);
                contributorVO.setOpenRank(openRank);
                contributorVO.setCooperation(cooperation.toString());
                contributorVO.setEffective(cooperation + Double.valueOf(openRank));
                contributorVO.setLink(cooperationUserNum);
                allCooperationMap.put(userId, cooperationUserNum);
                contributorVOS.add(contributorVO);
            }
        }, appThreadPool);

        // 获取此项目到查询时间的规模
        Double scale = 0.0;
        try {
            scale = CompletableFuture.supplyAsync(() -> {
                // 计算比例
                int nextTimeYear = nextTime.getYear();
                String nextTimeMonth = monthMap2Str.get(nextTime.getMonth().getValue());
                Map<String, Double> allProjectScale = getAllProjectScale(metaInfo.getPlatform(), nextMillis, nextTimeYear+"-"+nextTimeMonth);
                System.out.println("项目规模：" + allProjectScale);
                // 由小到大排序
                LinkedHashMap<String, Double> sortedMap = allProjectScale.entrySet().stream().sorted((v1, v2) -> v1.getValue().compareTo(v2.getValue())).collect(Collectors.toMap(
                        Map.Entry::getKey,
                        Map.Entry::getValue,
                        (e1, e2) -> e1,
                        LinkedHashMap::new
                ));
                // 设置最大比例
                Double MAX_SCALE = 0.05;
                // 设置最小比例
                Double MIN_SCALE = 0.01;
                // 计算比例
                Collection<Double> values = sortedMap.values();
                System.out.println(values.size() + " 个项目参与项目规模计算");
                Double[] array = values.toArray(new Double[values.size()]);
                return MIN_SCALE + (MAX_SCALE-MIN_SCALE)*((array[array.length-1] - sortedMap.get(metaInfo.getRepoId()))/(array[array.length-1]-array[0]));
            }, appThreadPool).get();
        } catch (Exception e) {
            System.out.println("项目规模计算出现异常");
            e.printStackTrace();
        }
        cooperationFuture.get();

        // 根据比例获得 关键贡献者列表
        Double num = contributorVOS.size() * scale;
        if (num < 5) {
            // 如果 5 个人都选不出来，就默认返回 5 个人
            num = 5.0;
        }
        ArrayList<ContributorVO> resVOS = new ArrayList<>();
        while (contributorVOS.size() > 0) {
            resVOS.add(contributorVOS.poll());
        }
        keyContributorsVO.setContributorList(resVOS);
        keyContributorsVO.setNum(num.intValue());

        // 对结果做归一化处理
//        normalization(keyContributorsVO);

        StringBuilder sb = new StringBuilder();
        sb.append("openSODA").append(":").append("contributor").append(":").append(name).append(":").append(date);
        stringRedisTemplate.opsForValue().set(sb.toString(), JSON.toJSONString(keyContributorsVO));
        return keyContributorsVO;
    }

    /**
     * 对结果做归一化处理
     */
    private void normalization(KeyContributorsVO keyContributorsVO) {
        // 取出其中的关键贡献者列表
        List<ContributorVO> contributorList = keyContributorsVO.getContributorList();
        // 取出 协作度 的最大和最小值
        Double cooperationMax = contributorList.stream().mapToDouble(d -> Double.parseDouble(d.getCooperation())).max().getAsDouble();
        Double cooperationMin = contributorList.stream().mapToDouble(d -> Double.parseDouble(d.getCooperation())).min().getAsDouble();
        // 取出 贡献度 的最大和最小值
        Double openRankMax = contributorList.stream().mapToDouble(d -> Double.parseDouble(d.getOpenRank())).max().getAsDouble();
        Double openRankMin = contributorList.stream().mapToDouble(d -> Double.parseDouble(d.getOpenRank())).min().getAsDouble();

        // (X-X_{min})/(X_{max}-X_{min})
        for (ContributorVO contributorVO : contributorList) {
            // 获取这个贡献者的 协作度
            Double cooperation = Double.valueOf(contributorVO.getCooperation());
            // 对 协作度 进行归一化
            Double cooperationRes = (cooperation - cooperationMin)/(cooperationMax-cooperationMin);
            contributorVO.setCooperation(cooperationRes.toString());
            // 获取这个贡献者的 贡献度
            Double openRank = Double.valueOf(contributorVO.getOpenRank());
            // 对 贡献度 进行归一化
            Double openRankRes = (openRank - openRankMin)/(openRankMax - openRankMin);
            contributorVO.setOpenRank(openRankRes.toString());

            // 注入最终影响力
            contributorVO.setEffective(cooperationRes + openRankRes);

        }
    }

    /**
     * 获取所有项目的项目规模
     */
    public Map<String, Double> getAllProjectScale(String platform, Long nextMillis, String dateStr) {
        HashMap<String, Double> res = new HashMap<>();
        platform = platform.toLowerCase();
        List<String> names = new ArrayList<>(){{add("pytorch/pytorch");add("nodejs/node");add("facebook/react");add("alibaba/nacos");}};
        for (String name : names) {
            CommunityOpenRankVO communityOpenRankVO = openRankServiceDomain.queryCommunityOpenRank(platform, name);
            String repoId = communityOpenRankVO.getMeta().getRepoId();
            // 获取代码行
            long codeLine = iotDbMybatisService.getCodeLine(repoId, dateStr);
//            long codeLine = 5000;
            // 获取贡献者数量
            int contributorCount = iotDbMybatisService.getContributorCount(repoId, nextMillis);
            // 获取 issue 数量
            int issueCount = iotDbMybatisService.getIssueCount(repoId, nextMillis);
            // 获取 pr 数量
            int prCount = iotDbMybatisService.getPrCount(repoId, nextMillis);
            // 获取 star 数量
            int starCount = iotDbMybatisService.getStarCount(repoId, nextMillis);
            // 获取 fork 数量
            int forkCount = iotDbMybatisService.getForkCount(repoId, nextMillis);
            // 项目规模
            Double projectScale = FormulaCoefficientEnum.W_8.getWeight()*codeLine + FormulaCoefficientEnum.W_9.getWeight()*contributorCount + FormulaCoefficientEnum.W_10.getWeight()*issueCount + FormulaCoefficientEnum.W_11.getWeight()*prCount + FormulaCoefficientEnum.W_12.getWeight()*starCount + FormulaCoefficientEnum.W_13.getWeight()*forkCount;
            res.put(repoId, projectScale);
        }
        return res;
    }

    /**
     * 协作度计算
     */
    private Double cooperationDegreeCalculation(String userId, List<String> userIds, ConcurrentHashMap<String, String> userOpenRankMap, ConcurrentHashMap<String, String> lastMonthUserOpenRankMap, Map<String, Map<String, Integer>> cooperationPrNumMap, Map<String, Map<String, Integer>> cooperationIssueNumMap, HashMap<String, String> cooperationUserNum) {
        // 协作度计算左半部分
        Double cooperationLeft = 0.0;
        // 协作度计算右半部分
        AtomicDouble cooperationRight = new AtomicDouble(0);

        // 获取总 openRank
        Double allOpenRank = allOpenRank(userIds, userOpenRankMap);
        // 遍历用户，查看和其他活跃用户之间的协作
        for (String othersId : userIds ) {
            // 如果是本人则跳过
            if (!othersId.equals(userId)) {
                // 根据公式进行计算
                // ① 计算 openRank 变化量
                String currentMonthOpenRank = userOpenRankMap.get("u" + othersId);
                String lastMonthOpenRank = lastMonthUserOpenRankMap.get("u" + othersId);
                Double openRankChange = lastMonthOpenRank != null ? Double.valueOf(currentMonthOpenRank)-Double.valueOf(lastMonthOpenRank) : Double.valueOf(currentMonthOpenRank);
                // 获取用户 openRank 占比
                double rate = Double.valueOf(currentMonthOpenRank) / allOpenRank;
                if (openRankChange > 0) {
                    // 获取用户 othersId 和用户 userId 在 pr、issue 中发生讨论的次数
                    Map<String, Integer> prNumMap = cooperationPrNumMap.get(othersId);
                    Integer cooperationPrNums = 0;
                    // pr 总的讨论次数
                    AtomicInteger prTotalNum = new AtomicInteger(0);
                    if (prNumMap != null) {
                        Integer num = prNumMap.get(userId);
                        cooperationPrNums = num != null ? num : 0;
                        cooperationRight.addAndGet(cooperationPrNums*rate);
                        if (!cooperationPrNums.equals(0)) {
                            prNumMap.values().forEach(prNum -> {
                                prTotalNum.addAndGet(prNum);
                            });
                        }
                    }

                    Map<String, Integer> issueNumMap = cooperationIssueNumMap.get(othersId);
                    Integer cooperationIssueNums = 0;
                    // issue 总的讨论次数
                    AtomicInteger issueTotalNum = new AtomicInteger(0);
                    if (issueNumMap != null) {
                        Integer num = issueNumMap.get(userId);
                        cooperationIssueNums = num != null ? num : 0;
                        cooperationRight.addAndGet(cooperationIssueNums*rate);
                        if (!cooperationIssueNums.equals(0)) {
                            issueNumMap.values().forEach(issueNum -> {
                                issueTotalNum.addAndGet(issueNum);
                            });
                        }
                    }
                    // 记录和用户 userId 有协作行为的 用户，及次数
                    cooperationUserNum.put(othersId, String.valueOf(cooperationPrNums + cooperationIssueNums));
                    // ② 算出比例
                    Double denominator = FormulaCoefficientEnum.W_4.getWeight()*prTotalNum.get() + FormulaCoefficientEnum.W_5.getWeight()*issueTotalNum.get();
                    Double y = 0.0;
                    if (!denominator.equals(0.0)) {
                        y = (FormulaCoefficientEnum.W_2.getWeight()*(cooperationPrNums) + FormulaCoefficientEnum.W_3.getWeight()*(cooperationIssueNums))/denominator;
                    }
                    // ③ 计算出用户 othersId 贡献度变化和用户 userId
                    cooperationLeft += y*openRankChange;
                }
            }
        }
        // 遍历所有 PR 和 Issue 中的评论，查看内容中是否有 @该用户 的情况，如果有就计数，即指标中的 【被提及数】
        Double cooperation = FormulaCoefficientEnum.W_6.getWeight()*cooperationLeft + FormulaCoefficientEnum.W_7.getWeight()*cooperationRight.get();
        return cooperation;
    }

    /**
     * 获取这个月用户总 OpenRank
     */
    public Double allOpenRank(List<String> userIds, ConcurrentHashMap<String, String> userOpenRankMap) {
        AtomicDouble atomicDouble = new AtomicDouble(0.0);
        for (String userId : userIds) {
            String currentMonthOpenRank = userOpenRankMap.get("u" + userId);
            atomicDouble.addAndGet(Double.valueOf(currentMonthOpenRank));
        }
        return atomicDouble.get();
    }

    /**
     * 按年来查
     * @param name
     * @param date
     * @return
     */
    private KeyContributorsVO queryKeyContributorListByYear(String name, String date) {
        String[] split = date.split("-");
        // 获取其中的年份
        String year = split[0];

        ConcurrentHashMap<String, String> userOpenRankMap = new ConcurrentHashMap<>();
        for (String month : months) {
            DataNode dataNode = (DataNode) ((Map<String, Object>)cacheManager.get("communityOpenRankCache")
                    .get(name))
                    .get(year + month);
            if (dataNode != null) {
                List<List<Double>> nodes = dataNode.getNodes();
                nodes.stream().forEach(n -> {
                    // 遍历所有节点，通过节点 id，在缓存中找到其 平台 id
                    String platformId = (String) cacheManager.get("nodeId").get(n.get(0).toString());
                    // 判断该 平台 id 是否是用户
                    if (platformId.charAt(0) == 'u') {

                        // 如果是用户则存入 userOpenRankMap
                        userOpenRankMap.put(platformId, String.valueOf(Double.valueOf(userOpenRankMap.getOrDefault(platformId, "0.0"))+Double.valueOf(n.get(2))));
                    }
                    // 如果不是则跳过
                });
            }
        }
        // 获取活跃用户在平台上的 id
        List<String> userList = userOpenRankMap.keySet().stream().map(key -> (String) key).collect(Collectors.toList());
        return null;
    }

    /**
     * 按总体来查
     * @param name
     * @param date
     * @return
     */
    private KeyContributorsVO queryKeyContributorListAll(String name, String date) {
        return null;
    }

}
